package cn.caplike.demo.project.reactor.quickstart;

import com.alibaba.fastjson.JSON;
import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

/**
 * Description: 简单创建 {@link Flux} 或 {@link Mono} 并订阅它们<br>
 * Reference: <a href="https://projectreactor.io/docs/core/release/reference/#_simple_ways_to_create_a_flux_or_mono_and_subscribe_to_it">Simple Ways to Create a Flux or Mono and Subscribe to It</a>
 *
 * @author LiKe
 * @version 1.0.0
 * @date 2020-08-20 15:08
 */
public class SimpleWays2CreateFluxOrMonoAndSubscribe {

    // ~ Simple ways to create flux or mono

    @Test
    public void testFluxCreating() {
        Flux<String> firstSequence = Flux.just("foo", "bar", "foobar");
        // 阻塞直到上游发出第一个值或是结束
        System.out.println("Block first: " + firstSequence.blockFirst()); // foo

        final String transformed = firstSequence.as(flux -> "transformed");
        System.out.println("Transformed: " + transformed);  //  transformed
    }

    @Test
    public void testFluxBuffer() {
        // ~ Operators that Change the Demand from Downstream
        final Flux<Integer> secondSequence = Flux.fromIterable(Arrays.asList(1, 2, 3));
        //   值得注意的是, 在 subscribe 那一层的请求, 可能被上游的运算符包装.
        //   典型的就是 buffer(N) 操作符: 如果 buffer(N) 这一层收到下游的 request(1) 请求, 实际上是请求的一个 buffer 的数据, 而不是 1 个数据.
        final Flux<List<Integer>> buffer = secondSequence.buffer(2);
        System.out.println("Buffered first: " + JSON.toJSONString(buffer.blockFirst())); // [1, 2]
        System.out.println("Buffered last: " + JSON.toJSONString(buffer.blockLast()));   // [3]

        //   还有一个名为 prefetch 的方法, 它接受 int 类型的参数, 这是一种能够修改下游请求的运算符.
        //      prefetch 如果不指定, 默认预抓取 32 个数据.
        //      补充优化, prefetch 的内部保证当有 75% 的预抓取请求被满足后, 当前层会向上游再请求 75% 的数据. 这是一个启发式的优化: 能够让这些运算符主动式的 "预料" 到即将到来的请求并及时响应.
    }

    @Test
    public void testMonoCreating() {
        Mono<String> firstSequence = Mono.empty();
        System.out.print("In 15s (If the provided timeout expires,a RuntimeException is thrown), the value is: ");
        System.out.println(firstSequence.block(Duration.ofSeconds(15)));

        final Mono<String> secondSequence = Mono.justOrEmpty("foo");
        System.out.println("The value: " + secondSequence.block());
    }

    // =================================================================================================================

    // ~ 4.3.1 Examples of subscribe()
    //   https://projectreactor.io/docs/core/release/reference/#_subscribe_method_examples

    @Test
    public void testFluxSubscription() {
        Flux<Integer> sequence = Flux.range(1, 4)
                .map(i -> {
                    if (i <= 3) return i;
                    throw new RuntimeException("Got to 4");
                });

        sequence.subscribe(
                System.out::println,
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Done"),
                // ~ Output
                //   1
                //   2
                //   3
                //   Error: java.lang.RuntimeException: Got to 4
                sub -> sub.request(4)
        );

        sequence.subscribe(
                System.out::println,
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Done"),
                // ~ Output
                //   1
                //   2
                //   3
                sub -> sub.request(3)
        );
    }

    @Test
    public void testMonoSubscription() {
        final Mono<String> sequence = Mono.justOrEmpty("caplike");

        // ~ 什么都不输出
        sequence.subscribe(
                value -> {
                    System.out.println("Produced value: " + value);
                },
                throwable -> {
                    // When error occurred
                    throw new RuntimeException(throwable);
                },
                () -> {
                    System.out.println("Run some code when the sequence successfully completes");
                },
                // Do something with the Subscription produced by this subscribe call
                Subscription::cancel
        );

        // ~ 输出
        //   Produced value: caplike
        //   Run some code when the sequence successfully completes
        sequence.subscribe(
                value -> {
                    System.out.println("Produced value: " + value);
                },
                throwable -> {
                    // When error occurred
                    throw new RuntimeException(throwable);
                },
                () -> {
                    System.out.println("Run some code when the sequence successfully completes");
                },
                subscription -> {
                    // Do something with the Subscription produced by this subscribe call
                    subscription.request(999L);
                }
        );

        sequence.subscribe(
                value -> {
                    System.out.println("Produced value: " + value);
                },
                throwable -> {
                    // When error occurred
                    throw new RuntimeException(throwable);
                },
                () -> {
                    System.out.println("Run some code when the sequence successfully completes");
                },
                subscription -> {
                    System.out.println("...");
                }
        );
    }

    // =================================================================================================================

    // ~ 4.3.2 Cancelling a subscribe() with Its Disposable
    //   https://projectreactor.io/docs/core/release/reference/#_cancelling_a_subscribe_with_its_disposable

    //   All these lambda-based variants of subscribe() have a Disposable return type.
    //   In this case, the Disposable interface represents the fact that the subscription can be cancelled, by calling its dispose() method.

    //   For a Flux or Mono, cancellation is a signal that the source should stop producing elements.
    //   However, it is NOT guaranteed to be immediate:
    //   Some sources might produce elements so fast that they could complete even before receiving the cancel instruction.

    //   Some utilities around Disposable are available in the Disposables class. Among these,
    //   Disposables.swap() creates a Disposable wrapper that lets you atomically cancel and replace a concrete Disposable.
    //   This can be useful, for instance, in a UI scenario where you want to cancel a request and replace it with a new one whenever the user clicks on a button.
    //   Disposing the wrapper itself closes it. Doing so disposes the current concrete value and all future attempted replacements.

    //   Another interesting utility is Disposables.composite(…​). This composite lets you collect several Disposable — for instance,
    //   multiple in-flight requests associated with a service call — and dispose all of them at once later on. Once the composite’s dispose() method has been called,
    //   any attempt to add another Disposable immediately disposes it.

    // =================================================================================================================

    // ~ 4.3.3 An Alternative to Lambdas: BaseSubscriber
    //   https://projectreactor.io/docs/core/release/reference/#_an_alternative_to_lambdas_basesubscriber

    @Test
    public void testBaseSubscriber() {
        final Flux<Integer> flux = Flux.range(1, 4);

        flux.subscribe(
                System.out::println,
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Done"),
                subscription -> subscription.request(10)
        );

        // -------------------------------------------------------------------------------------------------------------
        SampleSubscriber<Integer> subscriber = new SampleSubscriber<>();
        flux.subscribe(subscriber);
    }

    // BaseSubscriber 也提供了 requestUnbounded() 方法, 效果和 request(Long.MAX_VALUE) 一致, 和 cancel() 方法
    private static class SampleSubscriber<T> extends BaseSubscriber<T> {

        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribed");
            super.request(1);
        }

        @Override
        protected void hookOnNext(T value) {
            System.out.println(value);
            request(1);
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("Done");
        }
    }

    /**
     * Description: Backpressure 的其中一种体现形式为消费者/接收者通过向上游运算符请求 {@code request()}.
     *
     * @return void
     * @author LiKe
     * @date 2020-08-20 14:26:13
     */
    @Test
    public void testBaseSubscriberOnCancel() {
        // ~ 输出
        //   Request of 1
        //   Cancelling after having received 1
        Flux.range(1, 10)
                .doOnRequest(element -> System.out.println("Request of " + element))
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        // 在 request "上游" 之前, 响应式序列中并无数据流动
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        System.out.println("Cancelling after having received " + value);
                        cancel();
                    }
                });
    }

    // =================================================================================================================

    // ~ 4.3.4 On Backpressure and Ways to Reshape Requests
    //   https://projectreactor.io/docs/core/release/reference/#_on_backpressure_and_ways_to_reshape_requests

    // ~ limitRate 和 limitRequest
    //   limitRate(N) splits the downstream requests so that they are propagated upstream in smaller batches.
    //   For instance, a request of 100 made to limitRate(10) would result in, at most, 10 requests of 10 being propagated to the upstream.
    //   Note that, in this form, limitRate actually implements the replenishing optimization discussed earlier.

    //   The operator has a variant that also lets you tune the replenishing amount (referred to as the lowTide in the variant):
    //   limitRate(highTide, lowTide).
    //   Choosing a lowTide of 0 results in strict batches of highTide requests, instead of batches further reworked by the replenishing strategy.

    //   limitRequest(N), on the other hand, caps the downstream request to a maximum total demand.
    //   It adds up requests up to N. If a single request does not make the total demand overflow over N, that particular request is wholly propagated upstream.
    //   After that amount has been emitted by the source, limitRequest considers the sequence complete, sends an onComplete signal downstream, and cancels the source.
}
